package com.pingan.haofang.searchcloud.common.fetch;

import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import com.google.gson.JsonArray;
import com.google.gson.JsonElement;
import com.pingan.haofang.searchcloud.api.IndexRowData;
import com.pingan.haofang.searchcloud.api.matedata.IndexMeta;
import com.pingan.haofang.searchcloud.api.matedata.constant.MetadataFieldType;
import com.pingan.haofang.searchcloud.common.constants.QueryOptimizeEnum;
import com.pingan.haofang.searchcloud.common.dto.FetchDataRetryDTO;
import com.pingan.haofang.searchcloud.common.exception.FieldException;
import org.apache.commons.lang.StringUtils;

import java.util.ArrayList;
import java.util.List;

/**
 * @author SONGBANGYONG396
 * @date 2018/7/5
 * @since 1.0.0
 */
public abstract class AbstractStreamFetchIterator<E> implements IStreamFetchIterator<E> {

    private static final Gson GSON = new GsonBuilder().setDateFormat("yyyy-MM-dd HH:mm:ss").create();

    /**
     * 记录单次查询出现异常次数的初始化值
     */
    static final int INIT_INVOKE_COUNT_VALUE = 0;

    private FetchDataRetryDTO retryDTO;

    public FetchDataRetryDTO getRetryDTO() {
        return retryDTO;
    }

    /**
     * 抓取数据时每次查询允许出现最大异常次数
     */
    int singleQueryMaxRetryTime;

    /**
     * 抓取数据时所有查询累计允许出现最大异常次数
     */
    int allQueryMaxRetryTime;

    /**
     * 抓取数据时异常重试时间间隔（单位毫秒）
     */
    long retryTimeInterval;

    void setRetryDTO(FetchDataRetryDTO retryDTO) {
        this.retryDTO = retryDTO;
        this.singleQueryMaxRetryTime = retryDTO.getSingleQueryMaxRetryTime();
        this.allQueryMaxRetryTime = retryDTO.getAllQueryMaxRetryTime();
        this.retryTimeInterval = retryDTO.getRetryTimeInterval();
    }

    /**
     * 获取进度redis的key
     *
     * @param loopCount 仅在多线程RPC并开启查询优化时使用
     * @param total
     * @param pageNo
     * @param pageSize
     * @return
     */
    protected String getProgress(Integer loopCount, Long total, Integer pageNo, Integer pageSize, QueryOptimizeEnum
            queryOptimizeEnum) {

        if (QueryOptimizeEnum.ENABLE.equals(queryOptimizeEnum)) {
            return loopCount + "_" + total + "_" + pageNo + "_" + pageSize;
        } else {
            return loopCount + "_" + pageNo + "_" + pageSize;
        }

    }

    /**
     * 当前进度信息，用于日志
     *
     * @param total
     * @param pageNo
     * @param pageSize
     * @return
     */
    protected String getProgressInfo(Long total, Integer pageNo, Integer pageSize) {
        return " total:" + total + " pageNo:" + pageNo + " pageSize:" + pageSize;
    }

    /**
     * 当前进度信息，用于日志
     *
     * @param threadSeq
     * @param loopCount
     * @param total
     * @param pageNo
     * @param pageSize
     * @param lastMaxId
     * @return
     */
    protected String getProgressInfo(Integer threadSeq, Integer loopCount, Long total, Integer pageNo, Integer pageSize, Long lastMaxId) {
        return " threadSeq:" + threadSeq + " loopCount:" + loopCount + " total:" + total + " pageNo:" + pageNo + " " +
                "pageSize:" + pageSize + " lastMaxId:" + lastMaxId;
    }


    /**
     * 根据转换后的数据获取最大Id
     *
     * @param convertedDatas 转换后的数据
     * @param meta           元数据信息
     * @param <R>
     * @param <M>
     * @return
     */
    protected <R extends IndexRowData, M extends IndexMeta> Long getLastMaxId(List<R> convertedDatas, M meta,String indexField) {

        //最终字段名称
        final String realIndexFieldName;

        Long lastMaxId;

        MetadataFieldType fieldType = null;
        if (StringUtils.isNotEmpty(indexField)) {
            realIndexFieldName = indexField;
            if (meta.getFiledMetas().containsKey(indexField)) {
                fieldType = meta.getFiledMetas().get(indexField).getType();
            }
            if (null == fieldType) {
                throw new FieldException("unKnow column " + indexField + ",please check the schema field");
            }
        } else {
            realIndexFieldName = meta.getPk().getName();
            fieldType = meta.getPk().getType();
        }

        switch (fieldType) {
            case INT:
                lastMaxId = convertedDatas.stream()
                        .filter(e -> e != null && e.get(realIndexFieldName) != null)
                        .mapToLong(e -> Long.valueOf((Integer) e.get(realIndexFieldName))).max()
                        .orElseThrow(() -> new UnsupportedOperationException("indexField illegal" +
                                ",please change fetch data mode."));
                break;
            case LONG:
                lastMaxId = convertedDatas.stream().filter(e -> e != null && e.get(realIndexFieldName) != null)
                        .mapToLong(e -> (Long) e.get(realIndexFieldName)).max()
                        .orElseThrow(() -> new UnsupportedOperationException("indexField illegal" +
                                ",please change fetch data mode."));
                break;
            default:
                lastMaxId = null;
                break;

        }
        return lastMaxId;
    }

    /**
     * 根据未转换的数据获取lastMaxId
     *
     * @param datas
     * @param meta
     * @param <M>
     * @return
     */
    protected <T, M extends IndexMeta> Long getLastMaxIdForDatas(List<T> datas, M meta) {
        final Long lastMaxId;

        String pkFieldName = meta.getPk().getName();

        MetadataFieldType pkType = meta.getPk().getType();

        JsonArray array = GSON.toJsonTree(datas).getAsJsonArray();

        switch (pkType) {
            case INT:
                List<Integer> primaryKeys = new ArrayList<>(datas.size());
                for (JsonElement element : array) {
                    Integer primaryKey = element.getAsJsonObject().get(pkFieldName).getAsInt();
                    primaryKeys.add(primaryKey);
                }
                lastMaxId = primaryKeys.stream().max(Integer::compare).orElse(0).longValue();
                break;
            case LONG:
                List<Long> longPrimaryKeys = new ArrayList<>(datas.size());
                for (JsonElement element : array) {
                    Long primaryKey = element.getAsJsonObject().get(pkFieldName).getAsLong();
                    longPrimaryKeys.add(primaryKey);
                }
                lastMaxId = longPrimaryKeys.stream().max(Long::compare).orElse(0L);
                break;
            default:
                lastMaxId = null;
                break;
        }

        return lastMaxId;
    }


}
